package com.fengye.revoker;

import com.fengye.cluster.ClusterStrategy;
import com.fengye.cluster.engine.ClusterEngine;
import com.fengye.model.ProviderService;
import com.fengye.model.RemoteRequest;
import com.fengye.model.RemoteResponse;
import com.fengye.register.RegisterCenter;
import com.fengye.register.RegisterCenter4Invoker;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * 消费端 bean 代理工厂类
 * @Author fengye
 * @Date 2023/2/14 21:03
 **/
public class RevokerProxyBeanFactory implements InvocationHandler {

    /** 服务调用线程池 */
    private ThreadPoolExecutor fixedThreadPool = null;

    /** 服务接口 */
    private Class<?> targetInterface;

    /** 超时时间 */
    private int consumeTimeout;

    /** 服务调用方appKey */
    private String appKey;

    /** 调用者线程数 */
    private int threadWorkerNumber = 10;

    /** 负载均衡策略 */
    private String clusterStrategy;

    private static volatile RevokerProxyBeanFactory singleton;

    private RevokerProxyBeanFactory(Class<?> targetInterface, int consumeTimeout, String appKey, String clusterStrategy) {
        this.targetInterface = targetInterface;
        this.consumeTimeout = consumeTimeout;
        this.clusterStrategy = clusterStrategy;
        this.appKey = appKey;
    }

    /**
     * 获取单例实例对象
     *
     * @param targetInterface
     * @param consumeTimeout
     * @param appKey
     * @param clusterStrategy
     * @return
     */
    public static RevokerProxyBeanFactory getSingleton(Class<?> targetInterface, int consumeTimeout, String appKey, String clusterStrategy) {
        if(singleton == null) {
            synchronized (RevokerProxyBeanFactory.class) {
                if(singleton == null) {
                    singleton = new RevokerProxyBeanFactory(targetInterface, consumeTimeout, appKey,clusterStrategy);
                }
            }
        }
        return singleton;
    }

    /**
     * 获取代理对象
     *
     * @return
     */
    public Object getProxy() {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{targetInterface}, this);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 服务接口名称
        String serviceKey = targetInterface.getName();
        // 获取某个接口的服务提供者
        RegisterCenter4Invoker registerCenter4Invoker = RegisterCenter.singleton();
        List<ProviderService> providerServiceList = registerCenter4Invoker.getServiceMetaDataMap4Consume().get(serviceKey);
        // 根据软负载策略，从服务提供者列表选取本次调用服务提供者
        ClusterStrategy clusterStrategyService = ClusterEngine.queryClusterStrategy(clusterStrategy);
        ProviderService providerService = clusterStrategyService.select(providerServiceList);

        // 复制一份服务提供者信息
        ProviderService newProvider = providerService.clone();
        newProvider.setServiceMethod(method);
        newProvider.setServiceItf(targetInterface);

        // 声明调用RemoteRequest对象
        final RemoteRequest request = new RemoteRequest();
        // 本次调用的唯一标识，用于日志跟踪
        request.setUniqueKey(UUID.randomUUID().toString() + "-" + Thread.currentThread().getId());
        request.setProviderService(newProvider);
        request.setInvokeTimeout(consumeTimeout);
        request.setInvokedMethodName(method.getName());
        request.setAppKey(appKey);
        request.setArgs(args);
        try {
            // 构建发起调用的线程池
            if (fixedThreadPool == null) {
                synchronized (this) {
                    if (fixedThreadPool == null) {
                        fixedThreadPool = new ThreadPoolExecutor(threadWorkerNumber, threadWorkerNumber, 0L,
                                TimeUnit.MICROSECONDS, new LinkedBlockingQueue<>(128),
                                new ThreadFactoryBuilder().setNameFormat("invoker-pool-%d").build(),
                                new ThreadPoolExecutor.AbortPolicy());
                    }
                }
            }

            // 这里加一个重设机制，如果Provider调用失败，换一个？

            String serverIp = request.getProviderService().getServerIp();
            int serverPort = request.getProviderService().getServerPort();
            InetSocketAddress inetSocketAddress = new InetSocketAddress(serverIp, serverPort);

            Future<RemoteResponse> responseFuture = fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress, request));
            RemoteResponse response = responseFuture.get(request.getInvokeTimeout(), TimeUnit.MICROSECONDS);

            if (response == null || !response.isSuccess()) {
                throw new RuntimeException("service response is error.");
            }

            if (response.getResult() instanceof Throwable) {
                throw new RuntimeException((Throwable) response.getResult());
            }

            return response.getResult();
        } catch (Exception e) {
            throw new RuntimeException("service invoker is error.", e);
        }
    }
}
